package com.zlsf.publish;


import com.zlsf.entity.Media;
import com.zlsf.util.ConstStr;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.ext.web.RoutingContext;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class Subscribermanager {

  private static Subscribermanager instance;
  private final static ConcurrentHashMap<String, MessageConsumer> consumers = new ConcurrentHashMap<>();

  private Subscribermanager() {
  }

  public static Subscribermanager getInstance() {
    if (instance == null) {
      instance = new Subscribermanager();
    }
    return instance;
  }

  public void subscribe(String tag, RoutingContext context, Media.Type type, Vertx vertx) {
    if (type.equals(Media.Type.Video)) {
      log.info("channel: {} -> {}, subscriber: {}", Long.toHexString(hashCode() & 0xffffffffL), tag, context.request().remoteAddress());
      String router = tag + ConstStr.MEDIA_DATA_PUBLISH_ROUTER;
      if (!consumers.containsKey(router)) {
        PublishManager.getInstance().open(tag);
        this.regesterConsumer(router, context.vertx());
      }
      PublishManager.getInstance().subscribe(tag, type, context);
    }
  }


  /**
   * 订阅裸流
   */
  private void regesterConsumer(String router, Vertx vertx) {
    if (this.consumers.contains(router)) {
      return;
    }
    log.info("注册consumer：{}", router);
    //订阅发布的消息
    MessageConsumer<byte[]> flvConsumer = vertx.eventBus().consumer(router, msg -> {
      int sequence = Integer.parseInt(msg.headers().get("sequence"));
      long timestamp = Long.parseLong(msg.headers().get("timestamp"));
      int payloadType = Integer.parseInt(msg.headers().get("payloadType"));
      String mediaType = msg.headers().get("mediaType");
      String tag = msg.headers().get("tag");
      if (ConstStr.MEDIA_TYPE_VIDEO.equals(mediaType)) {
        PublishManager.getInstance().publishVideo(tag, sequence, timestamp, payloadType, msg.body());
      } else {
        PublishManager.getInstance().publishAudio(tag, sequence, timestamp, payloadType, msg.body());
      }
    });
    this.consumers.put(router, flvConsumer);
  }


  public void unsubscribe(String tag, RoutingContext ctx) {
    String router = tag + ConstStr.MEDIA_DATA_PUBLISH_ROUTER;
    if (consumers.containsKey(router)) {
      boolean canCloseConsumer = PublishManager.getInstance().unsubscribe(tag, ctx);
      if (canCloseConsumer) {
        log.info("注销consumer：{}", router);
        consumers.get(router).unregister();
        consumers.remove(router);
      }
    }

  }
}
